package queue

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/hibiken/asynq"
	"strings"
	"sync"
	"time"
)

var (
	mqConfigMux = &sync.Mutex{}
	mqConfigs   = map[string]*mqConfig{}

	mqClientMux = &sync.Mutex{}
	mqClients   = map[string]*asynq.Client{}

	mqRedisDefaultOpts = map[string]asynq.RedisClientOpt{
		"hk": {
			Addr:     "r-j6cclqcbxmucvhu4yf.redis.rds.aliyuncs.com:6379",
			Password: "HAZjm4dRW",
			DB:       1,
		},
		"sz": {
			Addr:     "r-wz92n0luac0fgzkrbb.redis.rds.aliyuncs.com:6379",
			Password: "HAZjm4dRW",
			DB:       1,
		},
		"devhk": {
			Addr:     "r-j6cclqcbxmucvhu4yf.redis.rds.aliyuncs.com:6379",
			Password: "HAZjm4dRW",
			DB:       2,
		},
		"devsz": {
			Addr:     "r-wz92n0luac0fgzkrbb.redis.rds.aliyuncs.com:6379",
			Password: "HAZjm4dRW",
			DB:       2,
		},
	}
)

var (
	ErrNonFailure = errors.New("non-failure")
	ErrSkipRetry  = asynq.SkipRetry
)

type mqConfig struct {
	MaxRetries     *int `json:"max_retries,omitempty"`
	Timeout        *int `json:"timeout,omitempty"`
	Concurrency    *int `json:"concurrency,omitempty"`
	Retention      *int `json:"retention,omitempty"`
	StrictPriority bool `json:"strict_priority,omitempty"`

	LogLevel *int `json:"log_level,omitempty"`

	RedisRegion    string `json:"redis_region,omitempty"`
	RedisAddress   string `json:"redis_address,omitempty"`
	RedisPassword  string `json:"redis_pass,omitempty"`
	RedisDB        int    `json:"redis_db,omitempty"`
	RedisClientOpt asynq.RedisClientOpt
}

func mqGetConfig(queueName string) *mqConfig {
	mqConfigMux.Lock()
	defer mqConfigMux.Unlock()

	if v, ok := mqConfigs[queueName]; ok {
		return v
	}

	ctx := context.Background()

	var defaultCfg *mqConfig
	err := g.Config().MustGet(ctx, "app.asynq").Struct(&defaultCfg)
	if err != nil {
		panic(fmt.Sprintf("failed to parse asynq global configuration 'app.asynq': %v", err))
	}

	var cfg *mqConfig
	err = g.Config().MustGet(ctx, fmt.Sprintf("app.asynq.%s", queueName), defaultCfg).Struct(&cfg)
	if err != nil {
		panic(fmt.Sprintf("failed to parse asynq configuration: 'app.asynq.%s', %v", queueName, err))
	}

	if cfg == nil {
		panic("invalid asynq configuration")
	}

	mqConfigs[queueName] = cfg

	for _, config := range []*mqConfig{cfg, defaultCfg} {
		if config != nil {
			if mqConfigs[queueName].MaxRetries == nil && config.MaxRetries != nil {
				mqConfigs[queueName].MaxRetries = config.MaxRetries
			}

			if mqConfigs[queueName].Timeout == nil && config.Timeout != nil {
				mqConfigs[queueName].Timeout = config.Timeout
			}

			if mqConfigs[queueName].Concurrency == nil && config.Concurrency != nil {
				mqConfigs[queueName].Concurrency = config.Concurrency
			}

			if mqConfigs[queueName].LogLevel == nil && config.LogLevel != nil {
				mqConfigs[queueName].LogLevel = config.LogLevel
			}

			if mqConfigs[queueName].Retention == nil && config.Retention != nil {
				mqConfigs[queueName].Retention = config.Retention
			}

			if mqConfigs[queueName].RedisClientOpt.Addr == "" {
				if config.RedisAddress != "" {
					mqConfigs[queueName].RedisClientOpt = asynq.RedisClientOpt{
						Addr:     config.RedisAddress,
						Password: config.RedisPassword,
						DB:       config.RedisDB,
					}
				} else {
					if v, ok := mqRedisDefaultOpts[strings.ToLower(config.RedisRegion)]; ok {
						mqConfigs[queueName].RedisClientOpt = v
					}
				}
			}
		}
	}

	if mqConfigs[queueName].RedisClientOpt.Addr == "" {
		panic(fmt.Sprintf("failed to parse queue configuration: invalid redis address for queue %s", queueName))
	}

	// 默认值
	if mqConfigs[queueName].MaxRetries == nil {
		mqConfigs[queueName].MaxRetries = helper.Ptr.IntPtr(10)
	}

	if mqConfigs[queueName].Timeout == nil {
		mqConfigs[queueName].Timeout = helper.Ptr.IntPtr(21600)
	}

	if mqConfigs[queueName].Concurrency == nil {
		mqConfigs[queueName].Concurrency = helper.Ptr.IntPtr(0)
	}

	if mqConfigs[queueName].LogLevel == nil {
		mqConfigs[queueName].LogLevel = helper.Ptr.IntPtr(4)
	}

	if mqConfigs[queueName].Retention == nil {
		mqConfigs[queueName].Retention = helper.Ptr.IntPtr(0)
	}

	// g.DumpWithType(mqConfigs)
	return mqConfigs[queueName]
}

func mqGetClient(queueName string, opt asynq.RedisClientOpt) *asynq.Client {
	mqClientMux.Lock()
	defer mqClientMux.Unlock()

	if v, ok := mqClients[queueName]; ok {
		return v
	}

	client := asynq.NewClient(opt)
	mqClients[queueName] = client
	return client
}

func mqGetHandlePattern(queueName string, channelName string) string {
	if channelName != "" {
		return channelName
	}
	return fmt.Sprintf("vip.aoscdn.com:%s", queueName)
}

func mqGetQueueName(queueName string, priority ...QueuePriority) string {
	p := QueuePriorityDefault
	if priority != nil && len(priority) > 0 {
		p = priority[0]
	}
	if p == QueuePriorityHigh {
		return fmt.Sprintf("%s:critical", queueName)
	}
	if p == QueuePriorityLow {
		return fmt.Sprintf("%s:low", queueName)
	}
	return fmt.Sprintf("%s:default", queueName)
}

func mqGetProduceOptions(queueName string, cfg *mqConfig, opts ...ProduceOption) (taskOpts []asynq.Option, priority QueuePriority) {
	priority = QueuePriorityDefault
	maxRetries := *cfg.MaxRetries
	timeout := *cfg.Timeout
	retention := *cfg.Retention

	for _, opt := range opts {
		switch opt := opt.(type) {
		case produceOptionProcessIn:
			taskOpts = append(taskOpts, asynq.ProcessIn(time.Duration(opt)*time.Second))
		case produceOptionPriority:
			priority = QueuePriority(opt)
		case produceOptionMaxRetries:
			maxRetries = int(opt)
		case produceOptionTimeout:
			timeout = int(opt)
		case produceOptionRetention:
			retention = int(opt)
		}
	}

	taskOpts = append(taskOpts, asynq.MaxRetry(maxRetries))

	taskOpts = append(taskOpts, asynq.Timeout(time.Duration(timeout)*time.Second))

	taskOpts = append(taskOpts, asynq.Queue(mqGetQueueName(queueName, priority)))

	taskOpts = append(taskOpts, asynq.Retention(time.Duration(retention)*time.Second))

	return taskOpts, priority
}

func mqGetConsumeOptions(queueName string, cfg *mqConfig, opts ...ConsumeOption) (config asynq.Config) {
	concurrency := *cfg.Concurrency
	strictPriority := cfg.StrictPriority

	for _, opt := range opts {
		switch opt := opt.(type) {
		case consumeOptionConcurrency:
			concurrency = int(opt)
		case consumeOptionStrictPriority:
			strictPriority = bool(opt)
		}
	}

	config = asynq.Config{
		LogLevel:    asynq.LogLevel(*cfg.LogLevel),
		Concurrency: concurrency,
		Queues: map[string]int{
			mqGetQueueName(queueName, QueuePriorityHigh):    6,
			mqGetQueueName(queueName, QueuePriorityDefault): 3,
			mqGetQueueName(queueName, QueuePriorityLow):     1,
		},
		IsFailure: func(err error) bool {
			return err != ErrNonFailure // If resource is not available, it's a non-failure error
		},
		StrictPriority: strictPriority,
	}

	return config
}

/*
	queueName: 队列名称
	channelName: 通道名称，如果为空，将使用queueName作为通道名称

	queue 和 channel 的关系
		1. 一个 queue 可以有多个 channel
		2. consume的时候，将只会消费queue下指定channel的消息

	这样一个服务，就只有一个队列，webui里面队列看起来也方便
	比如：
		easyQueue.Produce("wxtech", "gtp:gitmind", job)
		easyQueue.Produce("wxtech", "gtp:lightpdf", job)
	这样就是 wxtech 队列下面，再分别有 gtp:gitmind 和 gtp:lightpdf 两个通道

	gitmind的服务只需要消费 gtp:gitmind 通道的消息
		easyQueue.Consume("wxtech", "gtp:gitmind", func(job *easyQueue.QueueJob) error {})
	lightpdf的服务只需要消费 gtp:lightpdf 通道的消息
		easyQueue.Consume("wxtech", "gtp:lightpdf", func(job *easyQueue.QueueJob) error {})
*/
func Produce(queueName string, channelName string, job interface{}, opts ...ProduceOption) (jobId string, err error) {
	cfg := mqGetConfig(queueName)
	client := mqGetClient(queueName, cfg.RedisClientOpt)
	taskOpts, priority := mqGetProduceOptions(queueName, cfg, opts...)

	message := newQueueJob(job, priority)
	bytes, err := json.Marshal(message)
	if err != nil {
		return "", err
	}

	task := asynq.NewTask(
		mqGetHandlePattern(queueName, channelName),
		bytes,
		taskOpts...,
	)

	info, err := client.Enqueue(task)
	if err != nil {
		return "", err
	}
	return info.ID, err
}
func Consume(queueName string, channelName string, doJob func(job *QueueJob) error, opts ...ConsumeOption) {
	go func() {
		cfg := mqGetConfig(queueName)

		srv := asynq.NewServer(
			cfg.RedisClientOpt,
			mqGetConsumeOptions(queueName, cfg, opts...),
		)
		// mqMux maps a type to a handler
		srvMux := asynq.NewServeMux()
		srvMux.HandleFunc(mqGetHandlePattern(queueName, channelName), func(ctx context.Context, task *asynq.Task) error {
			// g.DumpWithType(ctx.Deadline())
			// return doJob(task.ResultWriter().TaskID(), task.Type(), task.Payload())

			c := make(chan error, 1)
			go func() {
				var job *QueueJob
				err := gconv.Struct(task.Payload(), &job)
				if err != nil {
					c <- err
					return
				}
				job.HandledAt = gtime.Now().TimestampMilli()
				job.ChannelName = task.Type()
				job.ID = task.ResultWriter().TaskID()

				c <- doJob(job)
			}()
			select {
			case <-ctx.Done():
				glog.Error(ctx, "cancellation signal received, abandon this work.")
				return ctx.Err()
			case res := <-c:
				return res
			}
		})

		if err := srv.Run(srvMux); err != nil {
			panic(fmt.Sprintf("failed to run easyQueue consumer: %v", err))
		}
	}()
}
func ConsumePro(queueName string, channelName string, useCoroutined uint, doJob func(job *QueueJob) error, opts ...ConsumeOption) {
	cfg := mqGetConfig(queueName)
	srv := asynq.NewServer(
		cfg.RedisClientOpt,
		mqGetConsumeOptions(queueName, cfg, opts...),
	)
	// mqMux maps a type to a handler
	srvMux := asynq.NewServeMux()
	srvMux.HandleFunc(mqGetHandlePattern(queueName, channelName), func(ctx context.Context, task *asynq.Task) error {
		// g.DumpWithType(ctx.Deadline())
		// return doJob(task.ResultWriter().TaskID(), task.Type(), task.Payload())
		c := make(chan error, 1)
		go func() {
			var job *QueueJob
			err := gconv.Struct(task.Payload(), &job)
			if err != nil {
				c <- err
				return
			}
			job.HandledAt = gtime.Now().TimestampMilli()
			job.ChannelName = task.Type()
			job.ID = task.ResultWriter().TaskID()
			c <- doJob(job)
		}()
		select {
		case <-ctx.Done():
			glog.Error(ctx, "cancellation signal received, abandon this work.")
			return ctx.Err()
		case res := <-c:
			return res
		}
	})
	if useCoroutined == 1 {
		go func() {
			if err := srv.Run(srvMux); err != nil {
				panic(fmt.Sprintf("failed to run easyQueue consumer: %v", err))
			}
		}()
	} else {
		if err := srv.Run(srvMux); err != nil {
			panic(fmt.Sprintf("failed to run easyQueue consumer: %v", err))
		}
	}
}
